package top.wshape1.rpc.consumer.handler;

import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import top.wshape1.rpc.common.constant.ChannelConstant;
import top.wshape1.rpc.message.PingPongMessage;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Wshape1
 * @version 1.0
 * @description
 * @since 2024-02-07
 */

@Slf4j
public class ConsumerIdleHandler extends ChannelDuplexHandler {
    private static final double IDLE_GAP = ChannelConstant.IDLE_WRITER_TIME * 1.5;

    private final AtomicInteger idleTimes = new AtomicInteger(0);

    private final AtomicLong lastIdleTime = new AtomicLong(0);

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            if (((IdleStateEvent) evt).state() == IdleState.WRITER_IDLE) {
                // 每 IDLE_WRITER_TIME s内无消息发送 就发送心跳信息
                try {
                    log.debug("发送一次心跳到{}", ctx.channel().remoteAddress());
                    ctx.writeAndFlush(PingPongMessage.Ping());
                } catch (Exception e) {
                    log.warn("Failed to send heartbeat. The connection was closed. ", e);
                    // 心跳发送失败，关闭channel
                    ctx.close();
                    return;
                }

                // 检测
                long nowIdleTime = System.currentTimeMillis();
                if (nowIdleTime - lastIdleTime.get() <= IDLE_GAP) {
                    // 两次发送心跳的间隔小于 IDLE_WRITER_TIME s，说明该channel 可能空闲
                    // 空闲次数+1
                    idleTimes.incrementAndGet();
                    log.debug("{}连续空闲次数：{}", ctx.channel().remoteAddress(), idleTimes);
                    if (idleTimes.get() >= ChannelConstant.MAX_IDLE_TIMES) {
                        // 空闲次数达到最大值，关闭channel
                        ctx.close();
                        return;
                    }
                }

                // 更新上次发送心跳的时间
                lastIdleTime.set(nowIdleTime);
            }
        }
    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if (!(msg instanceof PingPongMessage)) {
            resetLastIdleTime();
            resetIdleTimes();
        }
        super.write(ctx, msg, promise);
    }

    private void resetIdleTimes() {
        idleTimes.set(0);
    }

    private void resetLastIdleTime() {
        lastIdleTime.set(0);
    }

}
